package org.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.atguigu.gmall.realtime.app.BaseApp;
import org.atguigu.gmall.realtime.bean.TradeSkuOrderBean;
import org.atguigu.gmall.realtime.common.Constant;
import org.atguigu.gmall.realtime.function.AsyncDimFunction;
import org.atguigu.gmall.realtime.util.AtguiguUtil;
import org.atguigu.gmall.realtime.util.FlinkSinkUtil;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * Project:gmallRealTime
 * Package:org.atguigu.gmall.realtime.app.dws
 * Author: ZengHaiFeng
 * CreateTime:2023/5/5 18:15:50
 * Description: Todo
 * Version：1.0
 */
public class Dws_09_DwsTradeSkuOrderWindow_Cache_async extends BaseApp {
    public static void main(String[] args) {
        new Dws_09_DwsTradeSkuOrderWindow_Cache_async()
                .init(40010,2,
                        Constant.TOPIC_DWD_TRADE_ORDER_DETAIL,
                        "Dws_09_DwsTradeSkuOrderWindow_Cache_async");
    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> ds) {
        // parse date to pojo
        SingleOutputStreamOperator<TradeSkuOrderBean> parseStream = parseToPojo(ds);
        // remove duplicate data according to the order details id
        SingleOutputStreamOperator<TradeSkuOrderBean> distinctStream = distinctByOrderDetailId(parseStream);
        // window and aggregation
        SingleOutputStreamOperator<TradeSkuOrderBean> windowStream = windowAndAgg(distinctStream);
        // supplementary dimension
        SingleOutputStreamOperator<TradeSkuOrderBean> resultStream = joinDims(windowStream);

        // output to Clickhouse
        //resultStream.print();
        writeToClickhouse(resultStream);
    }

    private void writeToClickhouse(SingleOutputStreamOperator<TradeSkuOrderBean> resultStream) {
        resultStream.addSink(FlinkSinkUtil.getClickHouseSink("dws_trade_sku_order_window",
                TradeSkuOrderBean.class));
    }

    private SingleOutputStreamOperator<TradeSkuOrderBean> joinDims(SingleOutputStreamOperator<TradeSkuOrderBean> windowStream) {
        SingleOutputStreamOperator<TradeSkuOrderBean> skuInfoStream = AsyncDataStream
                .unorderedWait(windowStream, new AsyncDimFunction<TradeSkuOrderBean>() {
            @Override
            public String getTable() {
                return "dim_sku_info";
            }

            @Override
            public String getId(TradeSkuOrderBean bean) {
                return bean.getSkuId();
            }

            @Override
            public void addDim(TradeSkuOrderBean bean, JSONObject dim) {
                bean.setSkuName(dim.getString("SKU_NAME"));
                bean.setSpuId(dim.getString("SPU_ID"));
                bean.setTrademarkId(dim.getString("TM_ID"));
                bean.setCategory3Id(dim.getString("CATEGORY3_ID"));
            }
        }, 60, TimeUnit.SECONDS);
        SingleOutputStreamOperator<TradeSkuOrderBean> spuInfoStream = AsyncDataStream
                .unorderedWait(skuInfoStream, new AsyncDimFunction<TradeSkuOrderBean>() {
            @Override
            public String getTable() {
                return "dim_spu_info";
            }

            @Override
            public String getId(TradeSkuOrderBean bean) {
                return bean.getSpuId();
            }

            @Override
            public void addDim(TradeSkuOrderBean bean, JSONObject dim) {
                bean.setSpuName(dim.getString("SPU_NAME"));
            }
        }, 60, TimeUnit.SECONDS);
        SingleOutputStreamOperator<TradeSkuOrderBean> tmStream = AsyncDataStream
                .unorderedWait(spuInfoStream, new AsyncDimFunction<TradeSkuOrderBean>() {
                    @Override
                    public String getTable() {
                        return "dim_base_trademark";
                    }

                    @Override
                    public String getId(TradeSkuOrderBean bean) {
                        return bean.getTrademarkId();
                    }

                    @Override
                    public void addDim(TradeSkuOrderBean bean, JSONObject dim) {
                        bean.setTrademarkName(dim.getString("TM_NAME"));
                    }
                }, 60, TimeUnit.SECONDS);
        SingleOutputStreamOperator<TradeSkuOrderBean> c3Stream = AsyncDataStream
                .unorderedWait(tmStream, new AsyncDimFunction<TradeSkuOrderBean>() {
                    @Override
                    public String getTable() {
                        return "dim_base_category3";
                    }

                    @Override
                    public String getId(TradeSkuOrderBean bean) {
                        return bean.getCategory3Id();
                    }

                    @Override
                    public void addDim(TradeSkuOrderBean bean, JSONObject dim) {
                        bean.setCategory3Name(dim.getString("NAME"));
                        bean.setCategory2Id(dim.getString("CATEGORY2_ID"));
                    }
                }, 60, TimeUnit.SECONDS);
        SingleOutputStreamOperator<TradeSkuOrderBean> c2Stream = AsyncDataStream
                .unorderedWait(c3Stream, new AsyncDimFunction<TradeSkuOrderBean>() {
                    @Override
                    public String getTable() {
                        return "dim_base_category2";
                    }

                    @Override
                    public String getId(TradeSkuOrderBean bean) {
                        return bean.getCategory2Id();
                    }

                    @Override
                    public void addDim(TradeSkuOrderBean bean, JSONObject dim) {
                        bean.setCategory2Name(dim.getString("NAME"));
                        bean.setCategory1Id(dim.getString("CATEGORY1_ID"));
                    }
                }, 60, TimeUnit.SECONDS);
        return  AsyncDataStream
                .unorderedWait(c2Stream, new AsyncDimFunction<TradeSkuOrderBean>() {
                    @Override
                    public String getTable() {
                        return "dim_base_category1";
                    }

                    @Override
                    public String getId(TradeSkuOrderBean bean) {
                        return bean.getCategory1Id();
                    }

                    @Override
                    public void addDim(TradeSkuOrderBean bean, JSONObject dim) {
                        bean.setCategory1Name(dim.getString("NAME"));
                    }
                }, 60, TimeUnit.SECONDS);
    }

    private SingleOutputStreamOperator<TradeSkuOrderBean> windowAndAgg(SingleOutputStreamOperator<TradeSkuOrderBean> distinctStream) {
        return distinctStream.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<TradeSkuOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((bean,ts) -> bean.getTs())
                        .withIdleness(Duration.ofSeconds(60)))
                .keyBy(TradeSkuOrderBean::getSkuId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<TradeSkuOrderBean>() {
                    @Override
                    public TradeSkuOrderBean reduce(TradeSkuOrderBean value1, TradeSkuOrderBean value2) throws Exception {
                        value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                        value1.setOriginalAmount(value1.getOriginalAmount().add(value2.getOriginalAmount()));
                        value1.setActivityAmount(value1.getActivityAmount().add(value2.getActivityAmount()));
                        value1.setCouponAmount(value1.getCouponAmount().add(value2.getCouponAmount()));
                        return value1;
                    }
                }, new ProcessWindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                    @Override
                    public void process(
                            String s,
                            Context context,
                            Iterable<TradeSkuOrderBean> elements,
                            Collector<TradeSkuOrderBean> out) throws Exception {
                        TradeSkuOrderBean bean = elements.iterator().next();
                        bean.setStt(AtguiguUtil.tsToDateTime(context.window().getStart()));
                        bean.setEdt(AtguiguUtil.tsToDateTime(context.window().getEnd()));
                        bean.setTs(System.currentTimeMillis());
                        out.collect(bean);
                    }
                });
    }

    private SingleOutputStreamOperator<TradeSkuOrderBean> distinctByOrderDetailId(SingleOutputStreamOperator<TradeSkuOrderBean> parseStream) {
        return parseStream
                .keyBy(TradeSkuOrderBean::getOrderDetailId)
                .process(new KeyedProcessFunction<String, TradeSkuOrderBean, TradeSkuOrderBean>() {
                    private ValueState<TradeSkuOrderBean> lastBeanState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        lastBeanState = getRuntimeContext().getState(
                                new ValueStateDescriptor<>("lastBean", TradeSkuOrderBean.class)
                        );
                    }

                    @Override
                    public void processElement(
                            TradeSkuOrderBean value,
                            Context ctx,
                            Collector<TradeSkuOrderBean> out) throws Exception {
                        TradeSkuOrderBean lastBean = lastBeanState.value();
                        if (lastBean == null){
                            out.collect(value);
                        }else {
                            lastBean.setOriginalAmount(value.getOriginalAmount().subtract(lastBean.getOriginalAmount()));
                            lastBean.setCouponAmount(value.getCouponAmount().subtract(lastBean.getCouponAmount()));
                            lastBean.setActivityAmount(value.getActivityAmount().subtract(lastBean.getActivityAmount()));
                            lastBean.setOrderAmount(value.getOrderAmount().subtract(lastBean.getOrderAmount()));
                            out.collect(lastBean);
                        }
                        lastBeanState.update(value);
                    }
                });
    }

    private SingleOutputStreamOperator<TradeSkuOrderBean> parseToPojo(DataStreamSource<String> ds) {
        return ds
                .map(new MapFunction<String, TradeSkuOrderBean>() {
                    @Override
                    public TradeSkuOrderBean map(String value) {
                        JSONObject json = JSON.parseObject(value);
                        return TradeSkuOrderBean.builder()
                                .orderDetailId(json.getString("id"))
                                .skuId(json.getString("sku_id"))
                                .skuName(json.getString("sku_name"))
                                .originalAmount(json.getBigDecimal("split_original_amount"))
                                .activityAmount(json.getBigDecimal("split_activity_amount") == null ? new BigDecimal("0.0") : json.getBigDecimal("split_activity_amount"))
                                .couponAmount(json.getBigDecimal("split_coupon_amount") == null ? new BigDecimal("0.0") : json.getBigDecimal("split_coupon_amount"))
                                .orderAmount(json.getBigDecimal("split_total_amount"))
                                .ts(json.getLong("ts")*1000)
                                .build();
                    }
                });
    }
}
